package org.example.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.example.mapreduce.factory.CountJobFactory;
import org.example.mapreduce.factory.SortJobFactory;

import java.io.IOException;

/**
 * @author zhl
 * @version 1.0
 * @ClassName
 * @date 2020/11/10 15:11
 */
public class PhoneCount {

    private static Configuration conf;

    /**
     * 初始化参数信息，如果输入参数数量小于二则报错，如果大于二则将最后一个作为输出路径，
     * 其他的作为输入路径
     * */
    private static String[] init(String[] args) throws IOException {
        conf = new Configuration();
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        return otherArgs;
    }

    /**
     * 计算上行流量、下行流量分别的和，并计算总流量
     * */
    public static void countFlow(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String[] otherArgs = init(args);

        Job job = new CountJobFactory().getJob(conf);

        initFileInputFormat(job, otherArgs);
        initFileOutputFormat(job, otherArgs);

        System.exit(job.waitForCompletion(true)?0:1);
    }

    /**
     * 初始化输入流
     * */
    private static void initFileInputFormat(Job job, String[] otherArgs) throws IOException {
        for (int i = 0; i < otherArgs.length-1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
    }

    /**
     * 初始化输出流
     * */
    private static void initFileOutputFormat(Job job, String[] otherArgs) {
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
    }

    public static void sortByAllFlow(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String[] otherArgs = init(args);

        Job job = new SortJobFactory().getJob(conf);

        initFileInputFormat(job, otherArgs);
        initFileOutputFormat(job, otherArgs);
        System.exit(job.waitForCompletion(true)?0:1);
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//        PhoneCount.countFlow(args);
        PhoneCount.sortByAllFlow(args);
    }
}
